#include "thttpd.h"
#include <sys/types.h>
#include <sys/wait.h>
#include "shmem.h"
#include "msg_queue_peer.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <curl/curl.h>
#include "cJSON.h"
#include "MQTTClient.h"
#include <modbus/modbus.h>

#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ExampleClientPub"
#define TOPIC       "sensor/data/#"
#define QOS         1
#define TIMEOUT     10000L

struct shm_param para; // 共享内存的首地址
int *total;            // 指向共享内存中数据节点总个数
data_t *node_arr;      // 指向共享内存中节点缓存数组头
void *node_p;

// 从 node.json 读取 MQTT 服务器信息
int read_mqtt_info(const char *filename, char *addr, int *port) {
    FILE *fp = fopen(filename, "r");
    if (fp == NULL) {
        perror("Failed to open node.json");
        return -1;
    }

    fseek(fp, 0, SEEK_END);
    long size = ftell(fp);
    fseek(fp, 0, SEEK_SET);

    char *buffer = (char *)malloc(size + 1);
    if (buffer == NULL) {
        perror("Failed to allocate memory");
        fclose(fp);
        return -1;
    }

    fread(buffer, 1, size, fp);
    buffer[size] = '\0';
    fclose(fp);

    cJSON *root = cJSON_Parse(buffer);
    if (root == NULL) {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL) {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        free(buffer);
        return -1;
    }

    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    if (cJSON_IsObject(mqtt_server)) {
        cJSON *mqtt_addr = cJSON_GetObjectItem(mqtt_server, "addr");
        cJSON *mqtt_port = cJSON_GetObjectItem(mqtt_server, "port");
        if (cJSON_IsString(mqtt_addr) && cJSON_IsNumber(mqtt_port)) {
            strcpy(addr, mqtt_addr->valuestring);
            *port = mqtt_port->valueint;
        } else {
            cJSON_Delete(root);
            free(buffer);
            return -1;
        }
    } else {
        cJSON_Delete(root);
        free(buffer);
        return -1;
    }

    cJSON_Delete(root);
    free(buffer);
    return 0;
}

// 从 node.json 读取 Modbus 从机信息
int read_modbus_info(const char *filename, char *addr, int *port) {
    FILE *fp = fopen(filename, "r");
    if (fp == NULL) {
        perror("Failed to open node.json");
        return -1;
    }

    fseek(fp, 0, SEEK_END);
    long size = ftell(fp);
    fseek(fp, 0, SEEK_SET);

    char *buffer = (char *)malloc(size + 1);
    if (buffer == NULL) {
        perror("Failed to allocate memory");
        fclose(fp);
        return -1;
    }

    fread(buffer, 1, size, fp);
    buffer[size] = '\0';
    fclose(fp);

    cJSON *root = cJSON_Parse(buffer);
    if (root == NULL) {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL) {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        free(buffer);
        return -1;
    }

    cJSON *mb_dev = cJSON_GetObjectItem(root, "mb_dev");
    if (cJSON_IsObject(mb_dev)) {
        cJSON *mb_addr = cJSON_GetObjectItem(mb_dev, "addr");
        cJSON *mb_port = cJSON_GetObjectItem(mb_dev, "port");
        if (cJSON_IsString(mb_addr) && cJSON_IsNumber(mb_port)) {
            strcpy(addr, mb_addr->valuestring);
            *port = mb_port->valueint;
        } else {
            cJSON_Delete(root);
            free(buffer);
            return -1;
        }
    } else {
        cJSON_Delete(root);
        free(buffer);
        return -1;
    }

    cJSON_Delete(root);
    free(buffer);
    return 0;
}

// 连接 Modbus 从机
modbus_t* connect_modbus(const char *addr, int port) {
    modbus_t *ctx = modbus_new_tcp(addr, port);
    if (ctx == NULL) {
        fprintf(stderr, "Unable to allocate libmodbus context\n");
        return NULL;
    }

    if (modbus_connect(ctx) == -1) {
        fprintf(stderr, "Connection failed: %s\n", modbus_strerror(errno));
        modbus_free(ctx);
        return NULL;
    }

    printf("Connected to Modbus device %s:%d\n", addr, port);
    return ctx;
}

// 线程处理函数，并设置为非阻塞，自动回收线程资源
static void *msg_request(void *arg) {
    // 这里客户端描述符通过参数传进来了 accepted
    int sock = *((int *)arg);
    // 一般情况下，线程终止后，其终止状态一直保留到其它线程调用pthread_join获取它的状态为止。
    // 但是线程也可以被置为detach状态，这样的线程一旦终止就立刻回收它占用的所有资源，而不保留终止状态。
    pthread_detach(pthread_self());
    printf("进入线程函数\n");
    // handler_msg作为所有的请求处理入口
    return (void *)handler_msg(sock);
    // 将函数的返回值强转为void *类型的，实现线程函数返回值的统一
}

// MQTT 消息处理回调函数
int message_arrived(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    char *payload = (char *)message->payload;
    int type1, key, type2;
    float value;

    // 解析以空格分隔的消息
    if (sscanf(payload, "%d %d %d %f", &type1, &key, &type2, &value) == 4) {
        // 遍历共享内存节点，找到对应 key 并更新数据
        for (int i = 0; i < *total; i++) {
            if (node_arr[i].key == key) {
                switch (type2) {
                    case 1:
                        node_arr[i].new_val.b_val = (int)value;
                        break;
                    case 3:
                        node_arr[i].new_val.f_val = value;
                        break;
                    default:
                        break;
                }
                node_arr[i].ret = 0; // 标记采集成功
                break;
            }
        }
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 连接 MQTT 服务器并订阅主题
int connect_mqtt(const char *addr, int port) {
    char mqtt_address[100];
    snprintf(mqtt_address, sizeof(mqtt_address), "tcp://%s:%d", addr, port);

    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    MQTTClient_create(&client, mqtt_address, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, NULL, message_arrived, NULL);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %d\n", rc);
        return -1;
    }
    printf("Connected to MQTT server %s:%d\n", addr, port);

    if ((rc = MQTTClient_subscribe(client, TOPIC, QOS)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to subscribe, return code %d\n", rc);
        MQTTClient_disconnect(client, 10000);
        MQTTClient_destroy(&client);
        return -1;
    }

    // 保持 MQTT 连接
    while (1) {
        MQTTClient_yield();
    }

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}

/**
 * 这里是主函数
 * 可以用命令行参数自定义端口号
 */
int main(int argc, char *argv[]) {
    printf("开始运行.............\n");
    int ret = -1;

    ret = shm_init(&para, "222.c", MAX_NODE_SIZE);
    if (ret < 0) {
        return -1;
    }

    node_p = shm_getaddr(&para);
    if (node_p == NULL) {
        return -1;
    }
    printf("创建共享内存成功！\n");

    total = (int *)node_p;
    node_arr = (data_t *)((char *)node_p + sizeof(int));

    int port = 80;
    if (argc > 1) {
        port = atoi(argv[1]);
    }
    int lis_sock = init_server(port);

    // 读取 MQTT 信息并连接
    char mqtt_addr[100];
    int mqtt_port;
    if (read_mqtt_info("node.json", mqtt_addr, &mqtt_port) == 0) {
        // 创建线程处理 MQTT 连接
        pthread_t mqtt_thread;
        if (pthread_create(&mqtt_thread, NULL, (void *(*)(void *))connect_mqtt, (void *)&mqtt_port) != 0) {
            perror("Failed to create MQTT thread");
        }
    } else {
        printf("Failed to read MQTT info from node.json\n");
    }

    // 读取 Modbus 信息并连接
    char mb_addr[100];
    int mb_port;
    modbus_t *mb_ctx = NULL;
    if (read_modbus_info("node.json", mb_addr, &mb_port) == 0) {
        mb_ctx = connect_modbus(mb_addr, mb_port);
        if (mb_ctx == NULL) {
            printf("Failed to connect to Modbus device\n");
        }
    } else {
        printf("Failed to read Modbus info from node.json\n");
    }

    while (1) {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int sock = accept(lis_sock, (struct sockaddr *)&peer, &len);
        if (sock < 0) {
            perror("accept failed");
            continue;
        }

        int *new_sock = malloc(sizeof(int));
        if (new_sock == NULL) {
            perror("malloc failed");
            close(sock);
            continue;
        }
        *new_sock = sock;

        pthread_t tid;
        if (pthread_create(&tid, NULL, msg_request, (void *)new_sock) > 0) {
            perror("pthread_create failed");
            close(sock);
            free(new_sock);
        } else {
            printf("创建线程成功，sock = %d\n", sock);
        }
    }

    if (mb_ctx != NULL) {
        modbus_close(mb_ctx);
        modbus_free(mb_ctx);
    }
    return 0;
}
